package com.ccb.channel.pos;

import org.apache.log4j.Logger;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;

import com.alibaba.fastjson.JSON;
import com.ccb.filter.Filters;
import com.ccb.filter.RecordFilter;
import com.ccb.util.Record;

public class PosMapper extends Mapper<LongWritable, Text, Text, Text> {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = Logger.getLogger(PosMapper.class);

	private RecordFilter recordFilter ; 
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		recordFilter  =  Filters.getFilter("pos");
		if(recordFilter == null){
			logger.info("fiter is null");
		}
	}
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
		if (logger.isDebugEnabled()) {
			logger.debug("map(LongWritable, Text, Context) - start");
		}

	//	JsonObject json = new JsonParser().parse(value.toString()).getAsJsonObject();
	//	JSONObject json=JSONObject.parse(value.toString())
		Record rd = JSON.parseObject(value.toString(), Record.class);
		String serviceId = rd.getServiceId();
	
		if(recordFilter !=null && !recordFilter.accept(serviceId)){
			if (logger.isDebugEnabled()) {
				logger.debug("map(LongWritable, Text, Context) - end");
			}
			return;
		}
		// 配对key json
		context.write(new Text(rd.getUid()), value);

		if (logger.isDebugEnabled()) {
			logger.debug("map(LongWritable, Text, Context) - end");
		}
	}
}
